/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.ram;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.impl.ReadOnlyChannel;

/**
 * A read-only channel that reads data from several input channels. 
 * @author Alex Antonau
 * */
public class JoinChannel extends ReadOnlyChannel<Object[]> {

	private final Channel<?>[] inputs;

	public JoinChannel(Channel<?>... inputs) {
		super();
		if(inputs.length == 0) throw new IllegalArgumentException();
		this.inputs = inputs;
	}

	static final class ReadLock implements Lock<Object[]>{

		private final Lock<Object>[] inputLocks;
		
		public ReadLock(Lock<Object>[] inputLocks) {
			super();
			this.inputLocks = inputLocks;
		}

		@Override
		public int size() {
			return inputLocks[0].size();
		}

		@Override
		public Object[] get(int index) {
			Object[] res = new Object[inputLocks.length];
			for(int i=0;i<res.length;++i)res[i] = inputLocks[i].get(index);
			return res;
		}

		@Override
		public void set(int index, Object[] elem) {
			ReadOnlyChannel.raiseReadOnlyException();
		}

		@Override
		public Object[] toArray() {
			Object[] res = new Object[inputLocks.length];
			for(int i=0;i<res.length;++i)res[i] = inputLocks[i].toArray();
			return res;
		}

		@Override
		public void unlock() {
			for(Lock<?> l:inputLocks) l.unlock();
		}

		@Override
		public void cancel() {
			for(Lock<?> l:inputLocks) l.cancel();
		}
		
	}
	@SuppressWarnings("unchecked")
	@Override
	public Lock<Object[]> getReadLock(int size) {
		Lock<Object>[] inputLocks = new Lock[inputs.length];
		int i=0;
		while(i<inputLocks.length) {
			inputLocks[i] = (Lock<Object>) inputs[i].getReadLock(size);
			if(inputLocks[i] == null) break;
			++i;
		}
		if(i==inputLocks.length) return new ReadLock(inputLocks);
		while(i-->0)inputLocks[i].cancel();
		return null;
	}

	@Override
	public boolean isEmpty() {
		for(Channel<?> c : inputs) if(c.isEmpty()) return true;
		return false;
	}

	@Override
	public int size() {
		int size = Integer.MAX_VALUE;
		for(Channel<?> c : inputs) {
			final int cs = c.size();
			if(cs < size) size = cs;
			if(cs == 0) break;
		}
		return size;
	}

	@Override
	public void close() {
		for(Channel<?> c : inputs) c.close();
	}

	@Override
	public boolean isClosed() {
		for(Channel<?> c : inputs) if(!c.isClosed())return false;
		return true;
	}

	@Override
	public void fail(Throwable cause) {
		for(Channel<?> c : inputs) c.fail(cause);
	}

	@Override
	public boolean hasFailed() {
		for(Channel<?> c : inputs) if(c.hasFailed())return true;
		return false;
	}

	@Override
	public Throwable getFailureCause() {
		for(Channel<?> c : inputs) if(c.hasFailed())return c.getFailureCause();
		return null;
	}

	@SuppressWarnings("unchecked")
	@Override
	public Lock<Object[]> waitForReadLock(int size, long timeout) throws InterruptedException {
		Lock<Object>[] inputLocks = new Lock[inputs.length];
		long time=System.currentTimeMillis();
		int waitOnIndex = -1;
		do{
			int i=0;
			while(i<inputLocks.length) {
				if(waitOnIndex != i) inputLocks[i] = (Lock<Object>) inputs[i].getReadLock(size);
				if(inputLocks[i] == null) {
					if(inputs[i].isClosed() && inputs[i].size() < size)	waitOnIndex = -1;
					else waitOnIndex = i;
					break;
				}
				++i;
			}
			if(i == inputs.length) return new ReadLock(inputLocks);
			while(i-->0) inputLocks[i].cancel();
			
			if(waitOnIndex !=-1 && timeout > 0){
				inputLocks[waitOnIndex] = (Lock<Object>) inputs[waitOnIndex].waitForReadLock(size, timeout);
				timeout -= (System.currentTimeMillis()-time);
			}else break;
		} while (true);
		
		return null;
	}
	
}
